package cn.gwm.flink.streaming.strategy.vehiclepredict;

import cn.gwm.flink.streaming.constant.ChargePredictFields;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Author : GW00280745
 * Date : 2022/5/20 14:13
 * Description : 数据写入hive中
 * @author GW00280745
 */
public class ChargeData2Hive {
    public static void main(String[] args) {
        List<Object> keyListResult = ChargePredictFields.ins().resInfo.keySet().stream().sorted().collect(Collectors.toList());
        System.out.println(keyListResult);
    }
    public void process(){
        String path = ChargePredictFields.ins().confInfo.getOrDefault("hdfs.path", "");
        //textSink
        StreamingFileSink<String> hiveSink = hiveStreamingFileSink(path,null,".ext");
        List<Object> keyListResult = ChargePredictFields.ins().resInfo.keySet().stream().sorted().collect(Collectors.toList());
        System.out.println(keyListResult);
        batterIntervalStream
                //每条数据转成对应hive表的数据格式  需要先指定hive数据表
                .map(new Data2HiveFormat(keyListResult))
                .returns(Types.GENERIC(String.class))
                .addSink(hiveSink)
                .name("textFile2hive");
        //parquetSink
//        new ParquetFileSink().parquetFile2Hive(batterIntervalStream,path);
    }

    /**
     * @description: 获取hive sink类
     * @author:  GW00280745
     * @date: 2022/5/20 14:33
     * @param: outputPath
     * @param: partPrefix
     * @param: partSuffix
     * @return: org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink<java.lang.String>
     **/
    public StreamingFileSink<String> hiveStreamingFileSink(String outputPath, String partPrefix, String partSuffix){
//        OutputFileConfig config = OutputFileConfig
//                .builder()
//                //设置sink的前缀和后缀 默认是part开头的 所以采用默认的
//                .withPartPrefix(partPrefix)
//                //文件的头和文件扩展名 prefix-xxx-.ext
//                .withPartSuffix(partSuffix)
//                .build()
        //创建StreamingFileSink
        final StreamingFileSink<String> sink = StreamingFileSink
                .forRowFormat(new Path(outputPath),new SimpleStringEncoder<String>(ChargePredictFields.ins().utf8))
//                .withOutputFileConfig(config)
                /**
                 * 分桶策略
                 * DateTimeBucketAssigner --默认的桶分配政策，默认基于时间的分配器，每小时产生一个桶，格式如下yyyy-MM-dd--HH
                 * BasePathBucketAssigner ：将所有部分文件（part file）存储在基本路径中的分配器（单个全局桶）
                 */
//                .withBucketAssigner(new DateTimeBucketAssigner<>(EndTempConstant.dtYyyyMMdd))
                .withBucketAssigner(new ChargeBucketAssigner<>())
                /**
                 *  文件生成策略
                 *  CheckpointRollingPolicy
                 *  DefaultRollingPolicy
                 *  OnCheckpointRollingPolicy
                 */
                .withRollingPolicy(
                        /**
                         * Streaming File Sink 会将数据写入到桶中。由于输入流可能是无界的，因此每个桶中的数据被划分为多个有限大小的文件。
                         * 如何分桶是可以配置的，默认使用基于时间的分桶策略，这种策略每个小时创建一个新的桶，桶中包含的文件将记录所有该小时内从流中接收到的数据。
                         *
                         * 桶目录中的实际输出数据会被划分为多个部分文件（part file），每一个接收桶数据的 Sink Subtask ，至少包含一个部分文件（part file）。
                         * 额外的部分文件（part file）将根据滚动策略创建，滚动策略是可以配置的。默认的策略是根据文件大小和超时时间来滚动文件。超时时间指打开文件的最长持续时间，以及文件关闭前的最长非活动时间。
                         * 滚动策略决定了写出文件的状态变化过程
                         * 1. In-progress ：当前文件正在写入中
                         * 2. Pending ：当处于 In-progress 状态的文件关闭（closed）了，就变为 Pending 状态
                         * 3. Finished ：在成功的 Checkpoint 后，Pending 状态将变为 Finished 状态 需要设置checkpoint否则句柄不会释放
                         *
                         * 观察到的现象
                         * 1.会根据本地时间和时区，先创建桶目录
                         * 2.文件名称规则：part-<subtaskIndex>-<partFileIndex>
                         * 3.在macos中默认不显示隐藏文件，需要显示隐藏文件才能看到处于In-progress和Pending状态的文件，因为文件是按照.开头命名的
                         *
                         */
                        DefaultRollingPolicy.builder()
                                //滚动间隔 多长时间后生成一个文件 默认是60s。  它至少包含15分钟的数据
                                .withRolloverInterval(TimeUnit.MINUTES.toMillis(3))
                                //不活动间隔 多长时间没有写入就生成一个文件 默认是60s。  最近5分钟没有收到新的记录
                                .withInactivityInterval(TimeUnit.MINUTES.toMillis(1))
                                //最大尺寸 默认是128m。  文件大小达到 512m（写入最后一条记录后）
                                .withMaxPartSize(128*1024*1024L)
                                .build()
                )
                .build();
        return sink;
    }
    //动态属性设置

    private ChargeData2Hive(){}
    private ChargeData2Hive(Builder builder){
        this.batterIntervalStream = builder.batterIntervalStream;
    }
    private DataStream batterIntervalStream;
    public static class Builder{
        private DataStream batterIntervalStream;
        public Builder input(DataStream batterIntervalStream) {
            this.batterIntervalStream = batterIntervalStream;
            return this;
        }
        public ChargeData2Hive build(){
            return new ChargeData2Hive(this);
        }
    }
    public static Builder builder(){
        return new Builder();
    }
}
